package org.redis.cluster.redis.transport.codec.client;

import org.redis.cluster.redis.transport.RedisReply;
import org.redis.cluster.transport.ByteBuffer;
import org.redis.cluster.transport.protocol.ProtocolOutput;
import org.redis.cluster.transport.session.Session;
import org.redis.cluster.transport.support.AbstractReplayingProtocolDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RedisClientProtocolDecoder extends AbstractReplayingProtocolDecoder {
	
	private final static String START_BYTE="_start_byte";
	
	protected final Logger logger=LoggerFactory.getLogger(getClass());

	@Override
	protected void doDecode(ByteBuffer buffer, Session session, ProtocolOutput output) {
		if(!buffer.isReadable()) {
			return;
		}
		
		Byte startByte=session.getContext().get(START_BYTE);
		if(startByte==null) {
			startByte=buffer.readByte();
			session.getContext().put(START_BYTE, startByte);
			
			if(logger.isDebugEnabled()) {
				logger.debug("后端redis首字节 - {} - {}",startByte,(char) startByte.byteValue());
			}
		}
		
		switch (startByte) {
			case RedisReply.STATUS_START: {
				output(buffer.array(),session,output);
				break;
			}
			case RedisReply.ERROR_START: {
				output(buffer.array(),session,output);
				break;
			}
			case RedisReply.INTEGER_START: {
				output(buffer.array(),session,output);
				break;
			}
			case RedisReply.LENGTH_START: {
				int length=readInt(buffer);
				if(length==-1) {
					output(buffer.array(),session,output);
				}
				else {
					for(int i=0;i<length;i++) {
						buffer.readByte();
					}
					buffer.skip(2);
					output(buffer.array(),session,output);
				}
				break;
			}
			case RedisReply.COUNT_START: {
				int count=readInt(buffer);
				if(count!=0) {
					for(int i=0;i<count;i++) {
						byte lineStartByte=buffer.readByte();
						switch (lineStartByte) {
							case RedisReply.STATUS_START: {
								readLine(buffer);
								break;
							}
							case RedisReply.ERROR_START: {
								readLine(buffer);
								break;
							}
							case RedisReply.INTEGER_START: {
								readLine(buffer);
								break;
							}
							case RedisReply.LENGTH_START: {
								int length=readInt(buffer);
								if(length!=-1) {
									for(int j=0;j<length;j++) {
										buffer.readByte();
									}
									buffer.skip(2);
								}
								
								break;
							}
						}
					}
				}
				output(buffer.array(),session,output);
				break;
			}
		}
	}
	
	protected void output(byte value[],Session session, ProtocolOutput output) {
		session.getContext().remove(START_BYTE);
		output.out(session, value);
	}
	
	protected void readLine(ByteBuffer buffer) {
		byte temp;
		
		while((temp=buffer.readByte())!=RedisReply.END_1) {
			
		}
		buffer.skip(1);
	}
	
	protected int readInt(ByteBuffer buffer) {
		boolean isNegative=false;
		int result=0;
		byte temp;
		
		while((temp=buffer.readByte())!=RedisReply.END_1) {
			if(temp=='-') {
				isNegative=true;
				continue;
			}
			
			result=(result*10)+(temp-'0');
		}
		
		buffer.skip(1);
		
		if(isNegative) {
			return -result;
		}
		
		return result;
	}
}